package com.atguigu.project.app

import java.text.DecimalFormat

import com.atguigu.project.bean.UserVisitAction
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

/**
 * Author atguigu
 * Date 2020/11/2 15:13
 */
object PageConversionApp {
    def calcPageConversionRate(sc: SparkContext,
                               userVisitActionRDD: RDD[UserVisitAction],
                               pagesString: String) = {
        userVisitActionRDD.cache()
        // 1. 先把需要计算跳转率的页面id切出来
        val pages: Array[String] = pagesString.split(",")
        val prePages: Array[String] = pages.init // [1...6]
        val postPages: Array[String] = pages.tail // [2...7]
        
        // 1.1 做出来目标跳转流
        val targetFlows = prePages.zip(postPages).map {
            case (pre, post) => s"$pre->$post"
        }
        
        // 2. 计算分母  : 1-6这个6个页面的访问量
        val pageAndCount = userVisitActionRDD
            .filter(action => {
                prePages.contains(action.page_id.toString)
            })
            .map(action => (action.page_id, 1))
            .countByKey()
        // 3. 计算分子  "1->2"   "2->3"
        val pageFlows = userVisitActionRDD
            .groupBy(_.session_id)
            .flatMap {
                case (sid, it: Iterable[UserVisitAction]) =>
                    // 计算"1->"2多, "2->3"多少
                    val actionsSorted = it.toList.sortBy(_.action_time).map(_.page_id)
                    val preActions = actionsSorted.init
                    val postActions = actionsSorted.tail
                    preActions
                        .zip(postActions)
                        .map {
                            case (pre, post) => s"$pre->$post" // "1->2"   "2->3"
                        }
                        //.filter(flow => targetFlows.contains(flow))
                        .filter(targetFlows.contains)
            }
            .map((_, 1L))
            .countByKey()
        
        // 4. 计算跳转流
        val f = new DecimalFormat(".00%")
        val result = pageFlows.map {
            // "1->2"  1000
            case (flow, count) =>
                val molecular = count
                val denominator = pageAndCount(flow.split("->")(0).toLong) // 1
                (flow, f.format(molecular.toDouble / denominator))
        }
        //
        println(result)
        
    }
    
}

/*
计算分子:
    根据原始数据做出这样 RDD["1->2", '1->2', "...."]

    RDD[UserVisitAction] 安装sessionid进行分组
        .map(_.pageId)   // 1,2,...6

        .map(_.pageId)  // 2,3,...7

    做zip
        RDD[(1,2)...]
 */